package rabbitmq

import (
	"errors"
	"fmt"
	"gin/config/structs"
	"time"

	"github.com/panjf2000/ants"
	"github.com/sirupsen/logrus"
	"github.com/spf13/cast"
	"github.com/streadway/amqp"
)

// InitRabbitMQConsumer 初始化消费者
func (c *Connection) InitRabbitMQConsumer(isClose bool, rabbitMQConfig structs.RabbitMQConfig) {
	if isClose {
		c.CloseProcess <- true
	}
	c.Rabbit = rabbitMQConfig
	url := "amqp://" + c.Rabbit.UserName + ":" + c.Rabbit.PassWord + "@" + c.Rabbit.Host + ":" + cast.ToString(c.Rabbit.Port) + "/"
	//连接rabbit
	conn, err := amqp.Dial(url)
	if err != nil {
		logrus.Errorf("rabbit连接异常:%s", err.Error())
		logrus.Info("休息5S,开始重连rabbitMq消费者")
		time.Sleep(5 * time.Second)
		ants.Submit(func() { c.InitRabbitMQConsumer(false, c.Rabbit) })
		return
	}
	defer conn.Close()
	logrus.Info("rabbitMQ消费者连接成功")
	c.Conn = conn
	ch, err := conn.Channel()
	if err != nil {
		logrus.Info("RabbitMQ 开启通道错误: ", err.Error())
	}
	defer ch.Close()

	RabbitMqConn.GetMessage()
	// c.Conn = conn
	err = c.CreateRabbitMQConsumer()
	if err != nil {
		logrus.Error(err.Error())
		return
	}
	c.URL = url
	c.CloseProcess = make(chan bool, 1)
	c.ConsumerReConnect()
	logrus.Info("结束消费者旧主进程")
}

func (c *Connection) CreateRabbitMQConsumer() error {
	if len(c.RabbitConsumerList) == 0 {
		return errors.New("消费者信息不能为空")
	}
	var err error
	for _, value := range c.RabbitConsumerList {
		//创建一个通道
		c.Ch, err = c.Conn.Channel()
		if err != nil {
			return fmt.Errorf(fmt.Sprintf("MQ %s:%s", "打开Rabbit通道失败", err.Error()))
		}
		if err = c.Ch.ExchangeDeclare(
			value.ExchangeName,
			value.ExchangeType,
			true,
			false,
			false,
			false,
			nil,
		); err != nil {
			return fmt.Errorf(fmt.Sprintf("交换机初始化失败,交换机名称:%s,错误:%s", value.ExchangeName, err.Error()))
		}
		var queue amqp.Queue
		queue, err = c.Ch.QueueDeclare(
			// value.QueueName,
			"hello",
			true,
			false,
			false,
			false,
			nil,
		)
		if err != nil {
			return fmt.Errorf("队列初始化失败,队列名称:%s,错误: %s", value.QueueName, err.Error())
		}
		logrus.Info(fmt.Sprintf("declared Queue (%q %d messages, %d consumers), binding to Exchange (key %q)",
			queue.Name, queue.Messages, queue.Consumers, queue.Name))
		// 绑定队列
		err = c.Ch.QueueBind(value.QueueName, value.QueueName, value.ExchangeName, false, nil)
		if err != nil {
			return fmt.Errorf(fmt.Sprintf("MQ %s:%s", "绑定队列失败", err.Error()))
		}
		//绑定消费者
		messages := make(<-chan amqp.Delivery)
		messages, err = c.Ch.Consume(value.QueueName, "", false, false, false, false, nil)
		if err != nil {
			return fmt.Errorf(fmt.Sprintf("MQ %s:%s", "创建消费者失败", err.Error()))
		}
		ants.Submit(func() {
			c.ConsumeHandle(messages)
		})
	}
	return nil
}

// ConsumerReConnect 消费者重连
func (c *Connection) ConsumerReConnect() {
closeTag:
	for {
		c.ConnNotifyClose = c.Conn.NotifyClose(make(chan *amqp.Error))
		c.ChNotifyClose = c.Ch.NotifyClose(make(chan *amqp.Error))
		var err *amqp.Error
		select {
		case err = <-c.ConnNotifyClose:
			if err != nil {
				logrus.Errorf("rabbit消费者链接异常:%s", err.Error())
			}

		case err = <-c.ChNotifyClose:
			if err != nil {
				logrus.Errorf("rabbit消费者连接异常:%s", err.Error())
			}
			// 判断连接是否关闭
			if !c.Conn.IsClosed() {
				if err := c.Conn.Close(); err != nil {
					logrus.Errorf("rabbit连接关闭异常:%s", err.Error())
				}
			}
			_, isConnChannelOpen := <-c.ConnNotifyClose
			if isConnChannelOpen {
				close(c.ConnNotifyClose)
			}
			ants.Submit(func() {
				c.InitRabbitMQConsumer(false, c.Rabbit)
			})
			break closeTag
		case <-c.CloseProcess:
			break closeTag
		}
	}
	logrus.Info("结束消费者旧进程")
}

func (c *Connection) GetMessage() {
	ch, _ := c.Conn.Channel()
	defer ch.Close()

	// 声明要操作的队列
	q, _ := ch.QueueDeclare(

		"hello", // 队列名需要跟发送消息的队列名保持一致
		false,   // durable
		false,   // delete when unused
		false,   // exclusive
		false,   // no-wait
		nil,     // arguments
	)

	// 创建消息消费者
	msgs, _ := ch.Consume(
		q.Name, // 队列名
		"",     // 消费者名字，不填，则自动生成一个唯一ID
		true,   // 是否自动提交消息，即自动告诉rabbitmq消息已经处理成功。
		false,  // exclusive
		false,  // no-local
		false,  // no-wait
		nil,    // args
	)

	// 循环拉取队列中的消息
	for d := range msgs {
		// 打印消息内容
		// log.Printf("Received a message: %s", d.Body)
		// fmt.Println("Received a message:", d.Body)
		fmt.Println("zzzzzzz", string(d.Body))
	}
}
